package com.atguigu.cn.dataStream.api

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

import scala.util.Random

/**
 * @author yangshen
 * @date 2020/4/11 15:23
 */
//温度传感器读数样例类
case class SensorReading( id: String, timestamp: Long, temperature: Double )

object SourceTest {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    //1. 从自定义的集合中读取数据
    val stream1 = environment.fromCollection( List(
      SensorReading("sensor_1", 1547718199, 35.80018327300259),
      SensorReading("sensor_6", 1547718201, 15.402984393403084),
      SensorReading("sensor_7", 1547718202, 6.720945201171228),
      SensorReading("sensor_10", 1547718205, 38.101067604893444)
    ))

//    environment.fromElements(1, 2.0, "string").print()

    //2. 从文件中读取数据
    val stream2 = environment.readTextFile("D:\\my\\my_git\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    //3. 从kafka读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.16.26.16:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val stream3 = environment.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

    //4. 自定义source(测试环境用的比较多，生产环境用的少)
    val stream4 = environment.addSource(new SensorSource())

    //stream1.print("stream1").setParallelism(1)
    //stream2.print("stream2").setParallelism(1)
    //stream3.print("stream3").setParallelism(1)
    stream4.print("stream4").setParallelism(1)

    environment.execute("source test")

  }
}

class SensorSource() extends SourceFunction[SensorReading]{

  //定义一个flag，表示数据源是否正常运行
  var running: Boolean = true

  //取消数据源的生成
  override def cancel(): Unit = {
    running = false
  }

  //正常生成数据源
  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    //初始化一个随机数发生器
    val random = new Random()

    //初始化定义一组传感器温度数据
    var curTemp = 1.to(10).map(
      i => ("sensor_"+i, 60 + random.nextGaussian() * 20)
    )

    //用无线循环，产生数据流
    while (running) {

      //在前一次的温度基础上更新温度值
      curTemp = curTemp.map(
        t => (t._1, t._2 + random.nextGaussian())
      )

      //获取当前时间戳
      val curTime = System.currentTimeMillis()
      curTemp.foreach(
        //上下文收集并一条一条的输出
        t => sourceContext.collect( SensorReading(t._1, curTime,t._2) )
      )

      //设置时间间隔，方便观察输出
      Thread.sleep(500)
    }
  }



}


